package com.lili.collection.mq.config;

import cn.hutool.core.map.MapUtil;
import com.lili.collection.core.config.OutputConfig;
import com.lili.collection.core.domain.CollectCommonConfig;
import com.lili.collection.core.plugins.OutputPlugin;
import com.lili.collection.mq.plugins.KafkaOutputPlugin;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 *描述: kafka输出插件配置
 *@date: 2023/9/27  11:14
 *@author: lili
 */
@Slf4j
@Data
public class KafkaOutputConfig extends CollectCommonConfig implements OutputConfig {

    /**
     * 输出kafka配置
     */
    private Properties properties;

    /**
     * topic名称
     */
    private String topicName;

    @Override
    public OutputPlugin getPlugin() {
        return new KafkaOutputPlugin();
    }

    /**
     * 创建生产者配置
     * @return
     */
    public Properties createProducerProp(){
        if (MapUtil.isEmpty(this.properties)) {
            throw new IllegalArgumentException("kafka队列配置为空!");
        }
        if (!this.properties.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
            throw new IllegalArgumentException("kafka服务器地址不存在,初始化失败!");
        }
        if (StringUtils.isBlank(this.topicName)) {
            throw new IllegalArgumentException("kafka主题不能为空,初始化失败!");
        }
        if (!this.properties.containsKey(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)) {
            this.properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        }
        if (!this.properties.containsKey(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG)) {
            this.properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        }
        this.properties.put(ProducerConfig.RETRIES_CONFIG, 1);
        this.properties.put(ProducerConfig.ACKS_CONFIG, "all");
        return this.properties;
    }
}